package com.flink.test;

import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Description
 * @Author JL
 * @Date 2020/09/19
 * @Version V1.0
 */
public class DataTableToSet {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment sTableEnv = StreamTableEnvironment.create(sEnv);

        ExecutionEnvironment dEnv = ExecutionEnvironment.getExecutionEnvironment();
        BatchTableEnvironment tEnv = BatchTableEnvironment.create(dEnv);

        DataStream<Tuple2<Integer,Integer>> dateStream = sEnv.fromElements(Tuple2.of(1, 1), Tuple2.of(2, 0), Tuple2.of(3, 1));

        EnvironmentSettings bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
        TableEnvironment tableEnv = TableEnvironment.create(bbSettings);


        sTableEnv.createTemporaryView("temp_table", dateStream,  $("f0"), $("f1"));
        Table table = sTableEnv.sqlQuery("SELECT f0,f1 as f1 FROM temp_table");
        TableResult result = table.execute();
        result.print();

        TypeInformation<Tuple2<String, Integer>> info = TypeInformation.of(new TypeHint<Tuple2<String, Integer>>(){});
        DataSet<Tuple2<String, Integer>> dataSet = tEnv.toDataSet(table, info);
        dataSet.print();

        sEnv.execute("");
    }

}
